package com.shujia.card


import java.sql.{Connection, PreparedStatement}

import com.alibaba.fastjson.{JSON, JSONObject}
import com.shujia.card.RealTimeCardDayFlow.env
import com.shujia.flink.FlinkTool
import com.shujia.util.{CarUtil, JdbcUtil}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
  *
  * 1.2 按窗口统计每个卡扣流量 -
  * 统计每个卡扣最近5分钟车流量，每隔1分钟统计一次-  判断卡口拥堵情况
  */
object RealTimeCardWindowFlow extends FlinkTool {
  override def run(args: Array[String]): Unit = {

    //设置时间模式为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    /**
      * 获取过车数据
      *
      */

    val carsDS: DataStream[String] = CarUtil.loadKafkaData(env)


    /**
      * 取出卡口编号和时间字段
      *
      */

    val cardAndTimeDS: DataStream[(Long, Long)] = carsDS.map(car => {
      val carJson: JSONObject = JSON.parseObject(car)

      val card: Long = carJson.getLong("card")

      val time: Long = carJson.getLong("time")

      (card, time * 1000)
    })


    /**
      * 使用事件时间划分窗口
      * 1、指定时间模式为事件事件
      * 2、设置时间字段和水位线
      *
      */
    val assDS: DataStream[(Long, Long)] = cardAndTimeDS.assignTimestampsAndWatermarks(
      //指定时间戳字段和数据最大乱序时间
      new BoundedOutOfOrdernessTimestampExtractor[(Long, Long)](Time.seconds(10)) {
        override def extractTimestamp(element: (Long, Long)): Long = element._2
      }
    )

    /**
      * 划分窗口进行数据统计
      *
      */

    val sumFlowDS: DataStream[(Long, Long, Long)] = assDS
      .keyBy(_._1)
      .timeWindow(Time.minutes(5), Time.minutes(1))
      .process(new ProcessWindowFunction[(Long, Long), (Long, Long, Long), Long, TimeWindow] {
        override def process(key: Long,
                             context: Context,
                             elements: Iterable[(Long, Long)],
                             out: Collector[(Long, Long, Long)]): Unit = {

          //获取窗口的结束时间
          val windowEndTime: Long = context.window.getEnd

          //统计窗口内卡口的流量
          val size: Int = elements.size

          //将数据发送到下游
          out.collect((key, windowEndTime, size))

        }
      })

    /**
      * 将结果保存到mysql中
      *
      */

    sumFlowDS.addSink(new RichSinkFunction[(Long, Long, Long)] {


      var con: Connection = _

      override def invoke(value: (Long, Long, Long), context: SinkFunction.Context[_]): Unit = {

        val stat: PreparedStatement = con.prepareStatement("replace into real_time_card_window_flow(card,window_end_time,flow) values(?,?,?)")

        stat.setLong(1, value._1)
        stat.setLong(2, value._2)
        stat.setLong(3, value._3)

        stat.execute()

      }

      override def open(parameters: Configuration): Unit = {

        con = JdbcUtil.getConnection
      }

      override def close(): Unit = {
        con.close()
      }

    })

    env.execute("RealTimeCardWindowFlow$")


  }
}
